package run;

import lombok.extern.slf4j.Slf4j;
import org.springframework.messaging.rsocket.RSocketRequester;
import org.springframework.messaging.rsocket.annotation.ConnectMapping;
import org.springframework.stereotype.Controller;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;

import java.time.Duration;
import java.util.logging.Level;
import java.util.logging.Logger;

@Slf4j
@Controller
public class HelloController {
    // Must be initialized by @Test method...
//    volatile Sinks.Empty<Void> resultSink;
//    void reset() {
//        this.resultSink = Sinks.empty();
//    }
//
//    void await(Duration duration) {
//        this.resultSink.asMono().block(duration);
//    }

    private void runTest(Runnable testEcho) {

        Mono.fromRunnable(testEcho)
                .subscribeOn(Schedulers.boundedElastic()) // StepVerifier will block
                .subscribe(
                        aVoid -> {
                        },
                        ex -> {log.error("{}",ex.getMessage());},
                        () -> Mono.empty()
                );
    }


    @ConnectMapping("echo")
    public void echo(RSocketRequester requester) {
       runTest(() -> {
           // .onBackpressureBuffer()
            Flux<String> flux = Flux.range(1, 3).concatMap(i ->
                    requester.route("echo").data("Hello " + i).retrieveMono(String.class)).onBackpressureBuffer();

           StepVerifier.create(flux)
                   .expectNext("Hello 1")
                   .expectNext("Hello 2")
                   .expectNext("Hello 3")
                   .expectComplete().verify(Duration.ofSeconds(5));

        });
    }

}
